package cn.iotab.iotkit.transport.api.access.handler.impl;

import cn.iotab.iotkit.transport.api.access.handler.DeviceReplyListener;
import cn.iotab.iotkit.transport.api.access.model.message.DeviceMessage;
import cn.iotab.iotkit.transport.api.access.model.message.DeviceMessageReply;
import cn.iotab.iotkit.transport.api.common.id.DeviceMessageKey;
import com.google.common.util.concurrent.ListenableFuture;
import com.google.common.util.concurrent.SettableFuture;
import lombok.Getter;
import lombok.extern.slf4j.Slf4j;

import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.TimeoutException;

@Slf4j
public class DefaultDeviceReplyListener implements DeviceReplyListener {

    private long timeout = TimeUnit.MINUTES.toMillis(1); //默认1分钟超时

    private final Map<DeviceMessageKey, DeviceResponseMetaData> pendingRequest = new ConcurrentHashMap<>();

    /**
     * 运行一次超时检查
     */
    public void checkTimeout() {
        long now = System.currentTimeMillis();
        pendingRequest.forEach((key, value) -> {
            if (value.timeout > 0) {
                if (now - value.startTime > value.timeout) {
                    boolean found = onException(key, new TimeoutException());
                    if (found) {
                        log.trace("[{}] DeviceMessage Request timeout detected, timeout:{}", key, value.timeout);
                    }
                }
            }
        });
    }

    @Override
    public final void onReply(DeviceMessageReply reply) {
        DeviceMessageKey key = DeviceMessageKey.of(reply);
        DeviceResponseMetaData responseMetaData = pendingRequest.remove(key);
        if (responseMetaData != null) {
            log.trace("[{}] DeviceMessage Response received: {}", key, reply);
            responseMetaData.future.set(reply);
        } else {
            log.trace("[{}] DeviceMessage Response future not found, ignore response: {}", key, reply);
        }
    }

    public final boolean onException(DeviceMessageKey key, Throwable t) {
        DeviceResponseMetaData responseMetaData = pendingRequest.remove(key);
        if (responseMetaData != null) {
            log.trace("[{}] DeviceMessage request has an exception: {}", key, t.getClass().getName());
            responseMetaData.future.setException(t);
            return true;
        }
        return false;
    }

    public final ListenableFuture<DeviceMessageReply> getReply(DeviceMessage message) {
        DeviceMessageKey key = DeviceMessageKey.of(message);
        SettableFuture<DeviceMessageReply> future = SettableFuture.create();
        DeviceResponseMetaData metaData = new DeviceResponseMetaData(message, future, timeout);
        pendingRequest.put(key, metaData);
        return future;
    }


    @Getter
    private static class DeviceResponseMetaData {
        private final long startTime;
        private final long timeout;
        private final DeviceMessage message;
        private final SettableFuture<DeviceMessageReply> future;

        DeviceResponseMetaData(DeviceMessage message, SettableFuture<DeviceMessageReply> future, long timeout) {
            this.message = message;
            this.future = future;
            this.startTime = System.currentTimeMillis();
            this.timeout = timeout;
        }
    }
}
